package com.xb.system.service.impl;

import com.google.common.hash.BloomFilter;
import com.xb.model.domain.Users;
import com.xb.system.config.UserPhoneBloomFilterConfig;
import com.xb.system.service.MyDynamicDataSplitterService;
import com.xb.system.service.UsersService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 布隆过滤器去重，不建立唯一索引+Sharding-JDBC分表
 */
@Service
@Slf4j
public class MyDynamicDataSplitterServiceImpl implements MyDynamicDataSplitterService {

    private static final int BATCH_SIZE = 200000; // 每批处理的数据行数
    private static final int THREAD_POOL_SIZE = 10; // 线程池大小


    @Resource
    private UserPhoneBloomFilterConfig userPhoneBloomFilterConfig;
    @Resource
    private UsersService usersService;
    /**
     * 姓名脱敏
     *
     * @param name 原来的姓名
     * @return 脱敏后的姓名
     */
    private static String maskName(String name) {
        if (name.length() <= 1) {
            return name;
        }
        int length = name.length();
        StringBuilder maskedName = new StringBuilder();
        for (int i = 0; i < length - 1; i++) {
            maskedName.append("*");
        }
        maskedName.append(name.charAt(length - 1));
        return maskedName.toString();
    }

    @Override
    public void splitData(String filePath) throws IOException {
        BufferedReader reader = new BufferedReader(new InputStreamReader(Files.newInputStream(Paths.get(filePath))));
        ExecutorService executor = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
        BloomFilter<String> bloomFilter = userPhoneBloomFilterConfig.createBloomFilter();
        for (int i = 0; i < THREAD_POOL_SIZE; i++) {
            executor.submit(() -> processLines(reader,bloomFilter));
        }

        executor.shutdown();
        try {
            executor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            reader.close();
        }
    }

    private void processLines(BufferedReader reader, BloomFilter<String> userPhoneBloomFilter) {
        try {
            boolean endOfFile = false;
            // 文件末尾或所有行已处理
            do {
                List<Users> batch = new ArrayList<>();
                synchronized (reader) {
                    for (int i = 0; i < BATCH_SIZE; i++) {
                        String line = reader.readLine();
                        if (line == null) {
                            endOfFile = true;
                            log.info("已读完所有数据。");
                            break;
                        }
                        String[] parts = line.split("\\s+");
                        if (!parts[0].matches("^1[3456789]\\d{9}$")) {
                            continue;
                        }
                        if (parts.length >= 2) { // 只需要phone和name字段
                            parts[1] = maskName(parts[1]); // 对姓名进行脱敏
                            if (!userPhoneBloomFilter.mightContain(parts[0])) {
                                Users users = new Users(null, parts[0], parts[1]);
                                batch.add(users);
                                userPhoneBloomFilter.put(parts[0]);
                            }
                        }
                    }
                }
                if (!batch.isEmpty()) {
                    usersService.saveBatch(batch,BATCH_SIZE);
                }
            } while (!endOfFile);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    
}
